package com.example.flink;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
 import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * Flink任务启动器 - 暂时禁用
 */
@Slf4j
// @Component  // 临时注释掉，防止启动时加载
public class FlinkJobRunner implements ApplicationRunner {

    @Autowired
    private StreamExecutionEnvironment env;

    @Autowired
    private FinancialIndicatorJob financialIndicatorJob;

    @Autowired
    private CreditBehaviorJob creditBehaviorJob;

    @Value("${flink.remote.host}")
    private String flinkHost;

    @Value("${flink.remote.port}")
    private Integer flinkPort;

    private static final int MAX_RETRY_ATTEMPTS = 3;
    private static final long RETRY_DELAY_SECONDS = 10;

    @Override
    public void run(ApplicationArguments args) {
        log.info("开始初始化Flink任务...");
        log.info("正在连接远程Flink集群 {}:{}", flinkHost, flinkPort);

        if (env == null) {
            log.error("Flink执行环境未正确初始化");
            return;
        }

        // 启动Flink任务
        startJobWithRetry("财务指标", () -> {
            try {
                financialIndicatorJob.init();
                return true;
            } catch (Exception e) {
                log.error("财务指标Flink任务启动失败: {}", e.getMessage(), e);
                return false;
            }
        });

        startJobWithRetry("信贷行为", () -> {
            try {
                creditBehaviorJob.init();
                return true;
            } catch (Exception e) {
                log.error("信贷行为Flink任务启动失败: {}", e.getMessage(), e);
                return false;
            }
        });

        log.info("Flink任务初始化完成");
    }

    private void startJobWithRetry(String jobName, JobStarter jobStarter) {
        for (int attempt = 1; attempt <= MAX_RETRY_ATTEMPTS; attempt++) {
            log.info("第{}次尝试启动{}Flink任务", attempt, jobName);

            if (jobStarter.start()) {
                log.info("{}Flink任务启动成功", jobName);
                return;
            }

            if (attempt < MAX_RETRY_ATTEMPTS) {
                log.warn("{}Flink任务启动失败，{}秒后重试...", jobName, RETRY_DELAY_SECONDS);
                try {
                    TimeUnit.SECONDS.sleep(RETRY_DELAY_SECONDS);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error("重试等待被中断", e);
                    break;
                }
            } else {
                log.error("{}Flink任务启动失败，已达到最大重试次数", jobName);
            }
        }
    }

    @FunctionalInterface
    private interface JobStarter {
        boolean start();
    }
}